#include "../zhelpers.h"
#include "proto.msg.pb.h"
#include <iostream>
#include <fcntl.h>
#include <pthread.h>
using std::cout;
using std::cerr;
using std::endl;
using std::string;
#define THREAD_NUM 10
int read_image(const char *jpgfile, void *&buf)
{
	int fd = open( jpgfile, O_RDONLY );
	if(fd < 0){
		printf("Test image does not exist.\n");
		return -1;
	}
	struct stat  len_buf;
	fstat(fd, &len_buf);
	close(fd);
	int imageSize = len_buf.st_size;

	FILE *fp = fopen(jpgfile, "rb");
	buf = new unsigned char *[imageSize];
	fread(buf, 1, imageSize, fp);
	fclose(fp);
	return imageSize;
}

void *client_thread(void *context){
	int ret;
	//Socket to talk to server
	void *requester = zmq_socket(context, ZMQ_REQ);
	if(requester == NULL)
		return NULL;
	ret = zmq_connect(requester, "tcp://localhost:5559");
	if(ret == -1){
		cout << "connect error" << endl;
		return NULL;
	}
	int timeout = 60000;
	int zmq_hwm = 0;
	zmq_setsockopt(requester, ZMQ_SNDHWM, &zmq_hwm, sizeof(zmq_hwm));
	zmq_setsockopt(requester, ZMQ_SNDTIMEO, &timeout, sizeof(timeout));
	zmq_setsockopt(requester, ZMQ_RCVTIMEO, &timeout, sizeof(timeout));

	int request_nbr;
	char send_buf[128];
	for (request_nbr = 0; request_nbr != 100; request_nbr++) {
		void *image_buf;
		int image_size = read_image("10086.jpg", image_buf);
		if(image_size < 0){
			return NULL;
		}
		
		proto::front_req proto_msg;

		proto_msg.mutable_head()->set_version(420);
		proto_msg.mutable_head()->set_command(420);
		proto_msg.mutable_head()->set_password("xxxxxxxx", 8);
		proto_msg.mutable_head()->set_type(0);
		proto_msg.mutable_head()->set_status(0);
		proto_msg.mutable_head()->set_src(request_nbr);
		proto_msg.mutable_head()->set_dest(2);
		proto_msg.mutable_head()->set_sn_high(123456789);
		proto_msg.mutable_head()->set_sn_low(123456789);
		proto_msg.mutable_head()->set_length(99);

		proto_msg.set_class_id(70020);
		proto_msg.set_b2c_flag(0);
		proto_msg.set_cf_num(100);
		proto_msg.set_gf_num(100);
		proto_msg.set_lf_num(100);
		proto_msg.set_tv_num(100);
		proto_msg.set_image_size(image_size);
		proto_msg.set_image_data(image_buf, image_size);
		proto_msg.set_merge_same(false);

		string serial_stream = "";
		if(!proto_msg.SerializePartialToString(&serial_stream)){
			cerr << "fail to serial data" << endl;
			return NULL;
		}
//test the default value
/*
		proto::front_req msg2;
		if(!msg2.ParseFromString(serial_stream)){
			cerr << "ParseFromString fail" << endl;
			return -1;
		}
		cout << "head: " << msg2.head().has_version() << " " << msg2.head().version() << " " 
			<< msg2.head().has_command() << " " << msg2.head().command() << " "
			<< "body: " << msg2.has_class_id() << " " << msg2.class_id() << endl;
*/
//end
		
		int len = serial_stream.length();
		cout << "-------------------------------" << endl;
		cout << "start send num: " << request_nbr << "; size: " << len << endl;
		zmq_msg_t req_msg;
		zmq_msg_init_size(&req_msg, len);
		memcpy(zmq_msg_data(&req_msg), serial_stream.c_str(), len);
		int size = zmq_sendmsg(requester, &req_msg, 0);
		if(size == -1){
			cout << "send ERROR num: " << request_nbr << "; size: " << size << "; start recv" << endl;
			zmq_close(requester);
			requester = zmq_socket(context, ZMQ_REQ);
			if(requester == NULL){
				cout << "socket error" << endl;
				return NULL;
			}
			ret = zmq_connect(requester, "tcp://localhost:5559");
			if(ret == -1){
				cout << "connect error" << endl;
				return NULL;
			}
			zmq_setsockopt(requester, ZMQ_SNDHWM, &zmq_hwm, sizeof(zmq_hwm));
			zmq_setsockopt(requester, ZMQ_SNDTIMEO, &timeout, sizeof(timeout));
			zmq_setsockopt(requester, ZMQ_RCVTIMEO, &timeout, sizeof(timeout));

		}else
			cout << "send over num: " << request_nbr << "; size: " << size << "; start recv" << endl;


		//int zmq_setsockopt (void *socket, int option_name, const void *option_value, size_t option_len);

		//recv
		zmq_msg_t resp_msg;
		zmq_msg_init(&resp_msg);
		size = zmq_recvmsg(requester, &resp_msg, 0);
		if (size == -1){
			cout << "recv -1" << endl;
			continue;
		}
		cout << "recv over num: " << request_nbr << "; size: " << size << endl;
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&resp_msg), size);
		proto::front_resp front_resp;
		if(!front_resp.ParseFromString(proto_buf)){
			cout << "ParseFromString fail" << endl;
			continue;
		}
		cout << "get vsearch result, status: " << front_resp.head().status() << "; result num: " << front_resp.res_num() << endl;
		for(int i = 0; i < front_resp.res_num() && i < 5; i++){
			cout << "iid: " << front_resp.iid(i)
				<< "; gid: " << front_resp.gid(i)
				<< "; iid: " << front_resp.simi(i) << endl;
		}

		delete [] image_buf;
		image_buf = NULL;
	}
	zmq_close(requester);
}

int main(int argc, char **argv)
{
	timeval start,end;
	gettimeofday(&start, 0);
	void *context = zmq_ctx_new();
	int ret = 0;
	if(context == NULL)
		return -1;

	pthread_t tid[THREAD_NUM];

	for(int i = 0;i < THREAD_NUM;i++){
		pthread_create(&tid[i], NULL, client_thread, context);
	}
	
	for(int i = 0;i < THREAD_NUM;i++){
		pthread_join(tid[i], NULL);
	}

	zmq_ctx_destroy(context);
	gettimeofday(&end, 0);

	cout << "time: " << (end.tv_usec - start.tv_usec)/1000 + (end.tv_sec - start.tv_sec)*1000 << " ms" << endl;
	return 0;
}



